/*
 * brin_bloom.c
 *		BRIN 的 Bloom 操作类的实现
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * 一个将页面范围汇总到 Bloom 过滤器中的 BRIN 操作类。
 *
 * Bloom 过滤器允许有效测试给定页面范围是否包含
 * 特定值。因此，如果我们将每个页面范围汇总到一个小型
 * Bloom 过滤器中，我们可以轻松（且便宜）地测试它是否
 * 包含我们稍后获得的值。
 *
 * 该索引仅支持等值操作符，类似于哈希索引。
 * 但是，Bloom 索引要小得多，仅支持位图扫描。
 *
 * 注意：不要将此与在贡献模块中实现的 Bloom 索引混淆。
 * 该扩展实现了一个完全新的 AM，在单行的多个列上构建
 * 一个 Bloom 过滤器。此操作类与现有的 AM（BRIN）一起
 * 工作，并在一列上构建 Bloom 过滤器。
 *
 *
 * 值与哈希
 * -----------------
 *
 * 原始列值不会直接使用，而是首先使用
 * 常规类型特定的哈希函数进行哈希，生成 uint32 哈希。
 * 然后将此哈希值添加到摘要中——即再次进行哈希
 * 并添加到 Bloom 过滤器中。
 *
 * 这使得代码可以以相同的方式处理所有数据类型
 * （byval/byref/...），只需最小的空间要求，因为我们处理的是哈希
 * 而不是原始值。一切都是 uint32。
 *
 * 当然，这假设内置的哈希函数相当好，没有太多冲突等。
 * 但根据过去的经验，这似乎确实是事实。毕竟，对于哈希索引、
 * 哈希分区等使用相同哈希函数。
 *
 *
 * 哈希方案
 * --------------
 *
 * Bloom 过滤器需要多个独立的哈希函数。构造它们的
 * 方案有多种——例如，我们可以使用随机种子的
 * hash_uint32_extended，但这似乎非常昂贵。
 * 我们使用一种只需要两个函数的方案，如下所述：
 *
 * 更少哈希，相同性能：构建更好的 Bloom 过滤器
 * Adam Kirsch, Michael Mitzenmacher, 哈佛工程与应用科学学院,
 * 马萨诸塞州剑桥 [DOI 10.1002/rsa.20208]
 *
 * 两个哈希函数 h1 和 h2 使用硬编码的种子计算，
 * 然后使用 (h1 + i * h2) 组合以生成哈希函数。
 *
 *
 * 调整 Bloom 过滤器大小
 * -----------------------
 *
 * Bloom 过滤器的大小取决于我们将存储在
 * 其中的不同值的数量以及所需的假阳性率。
 * 不同值的数量越多和/或假阳性率越低，
 * Bloom 过滤器就越大。另一方面，我们希望尽可能
 * 将索引保持较小——这是 BRIN 索引的基本优点之一。
 *
 * 尽管不同元素的数量（在一个页面范围内）取决于
 * 数据，但我们可以将其视为固定的。这简化了
 * 假阳性率与大小之间的权衡。
 *
 * 在页面范围级别，假阳性率是 Bloom 过滤器匹配
 * 随机值的概率。对于整个索引（具有足够的
 * 页面范围），它代表索引范围的部分（因此
 * 需扫描的表的部分）与随机值匹配。
 *
 * 此外，Bloom 过滤器的大小受到实现
 * 限制——它必须适合一个索引页面（默认 8kB）。
 * 由于位图固有地是随机的（当“满”时大约一半的位
 * 被设置为 1，随机），压缩的帮助不会很大。
 *
 * 为了减小过滤器的大小（以适应页面），我们必须
 * 或者接受更高的假阳性率（不受欢迎），或者减少
 * 要存储在过滤器中的不同项的数量。我们当然不能
 * 改变输入数据，但我们可以使 BRIN 页面的范围更小
 * ——与默认的 128 个页面（1MB）相比，我们可以构建
 * 16 页范围的索引，或者类似的东西。这应该减少页面
 * 范围中的不同值的数量，从而使过滤器更小
 * （在假阳性率固定的情况下）。即使对于随机数据集
 * 这也应该有所帮助，因为每个堆页面中的行数是有限的
 * （对于非常窄的表大约是 ~290，实际上大约是 ~20）。
 *
 * 当然，好的大小决策依赖于拥有必要的数据，
 * 即页面范围内（给定大小）的不同值的数量和
 * 表的大小（以估算由于具有更大索引与扫描更大索引
 * 之间的假阳性率变化带来的成本变化）。我们可能
 * 没有这些数据——例如，在空表上构建索引时
 * 实际上无法做到这一点。而对于某些数据，我们只有
 * 整个表的估算，并且只能估算每个范围的值（ndistinct）。
 *
 * 另一个挑战是，虽然 Bloom 过滤器是按列的，
 * 但是整个索引元组必须适合一个页面。对于多列
 * 索引，这可能包括我们无法控制的部分（不一定是
 * Bloom 过滤器，其他列可能会使用其他 BRIN 操作类）。
 * 因此，如何在这些列之间分配空间并不是完全明确。
 *
 * 当前逻辑，实现在 brin_bloom_get_ndistinct 中，尝试根据
 * BRIN 范围的大小和每个范围的最大行数做出一些基本的大小
 * 决策。
 *
 *
 * IDENTIFICATION
 *	  src/backend/access/brin/brin_bloom.c
 */
#include "postgres.h"

#include "access/genam.h"
#include "access/brin.h"
#include "access/brin_internal.h"
#include "access/brin_page.h"
#include "access/brin_tuple.h"
#include "access/hash.h"
#include "access/htup_details.h"
#include "access/reloptions.h"
#include "access/stratnum.h"
#include "catalog/pg_type.h"
#include "catalog/pg_amop.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/syscache.h"

#include <math.h>

#define BloomEqualStrategyNumber	1

/*
 * 附加的 SQL 级支持函数。我们只有一个，它用于
 * 计算输入值的哈希。
 *
 * 过程编号不得使用保留给 BRIN 本身的值；请参见
 * brin_internal.h。
 */
#define		BLOOM_MAX_PROCNUMS		1	/* 我们需要的最大支持过程 */
#define		PROCNUM_HASH			11	/* 要求 */

/*
 * 从 procnum 中减去这个值以获得 BloomOpaque 数组中的索引
 * （必须等于私有 procnums 的最小值）。
 */
#define		PROCNUM_BASE			11

/*
 * BRIN 的 reloptions 的存储类型。
 */
typedef struct BloomOptions
{
	int32		vl_len_;		/* varlena 头部（请勿直接触摸！） */
	double		nDistinctPerRange;	/* 每个范围的不同值的数量 */
	double		falsePositiveRate;	/* 布隆过滤器的假阳性 */
} BloomOptions;

/*
 * 当前的最小值（16）是相当随意的，但它基于
 * 过滤器头部约为 20B，这大约是具有 1% 假阳性率的 16 个不同项
 * 的过滤器位图的大小。因此，允许较小的值我们不会获得太多好处。
 * 在任何情况下，最小值都不应大于 MaxHeapTuplesPerPage
 * （~290），这是单页范围的理论最大值。
 */
#define		BLOOM_MIN_NDISTINCT_PER_RANGE		16

/*
 * 用于根据页面范围内的行数确定不同项的数量。
 * 10% 的值在某种程度上类似于 estimate_num_groups
 * 的功能，因此我们在这里使用相同的因子。
 */
#define		BLOOM_DEFAULT_NDISTINCT_PER_RANGE	-0.1	/* 10% 的值 */

/*
 * 允许的假阳性范围及其默认值。确切的
 * 值在某种程度上是随意选择的，但考虑到了各种
 * 参数（过滤器大小与页面大小等）。
 *
 * 假阳性率越低，过滤器越精确，但
 * 其大小也会增加 — 在某些情况下，这会消除 BRIN 索引的
 * 主要优点，即小巧的大小。在 0.01% 时，索引大约为
 * 表的 10%（假设每个 8kB 页面有 290 个不同值）。
 *
 * 另一方面，随着假阳性率的增加，由于不匹配的
 * 原因，需要扫描表的更大部分 — 在 25% 时，我们可能
 * 接近顺序扫描更便宜的情况。
 */
#define		BLOOM_MIN_FALSE_POSITIVE_RATE	0.0001	/* 0.01% fp 率 */
#define		BLOOM_MAX_FALSE_POSITIVE_RATE	0.25	/* 25% fp 率 */
#define		BLOOM_DEFAULT_FALSE_POSITIVE_RATE	0.01	/* 1% fp 率 */

#define BloomGetNDistinctPerRange(opts) \
	((opts) && (((BloomOptions *) (opts))->nDistinctPerRange != 0) ? \
	 (((BloomOptions *) (opts))->nDistinctPerRange) : \
	 BLOOM_DEFAULT_NDISTINCT_PER_RANGE)

#define BloomGetFalsePositiveRate(opts) \
	((opts) && (((BloomOptions *) (opts))->falsePositiveRate != 0.0) ? \
	 (((BloomOptions *) (opts))->falsePositiveRate) : \
	 BLOOM_DEFAULT_FALSE_POSITIVE_RATE)

/*
 * 估计我们可以放在一页上最大的布隆。这并不是
 * 一个完美的保证，原因有几个。例如，行可能
 * 更大，因为索引有多个列。
 */
#define BloomMaxFilterSize \
	MAXALIGN_DOWN(BLCKSZ - \
				  (MAXALIGN(SizeOfPageHeaderData + \
							sizeof(ItemIdData)) + \
				   MAXALIGN(sizeof(BrinSpecialSpace)) + \
				   SizeOfBrinTuple))

/*
 * 用于计算两个哈希函数 h1 和 h2 的种子，然后用于
 * 使用 (h1 + i * h2) 方案生成 k 个哈希。
 */
#define BLOOM_SEED_1	0x71d924af
#define BLOOM_SEED_2	0xba48b314

/*
 * Bloom Filter
 *
 * 表示一个布隆过滤器，基于索引值的哈希构建。也就是说，
 * 我们计算该值的 uint32 哈希，然后将这个哈希存入
 * 布隆过滤器中（并对其计算其他哈希）。
 *
 * XXX 我们可以实现“稀疏”布隆过滤器，仅保留那些
 * 并不是全为 0 的字节。但尽管索引不支持 TOAST，varlena
 * 仍然可以被压缩。所以这似乎是多余的，因为压缩
 * 应该做同样的事情。
 *
 * XXX 我们还可以监控布隆过滤器中设置的位数，然后
 * 在假阳性率过高时停止使用它（并不存储位图，以节省空间）。
 * 但即使假阳性率超过了预期值，它仍然可以消除一些页面范围。
 */
typedef struct BloomFilter
{
	/* varlena 头部（请勿直接触摸！） */
	int32		vl_len_;

	/* 各种标志的空间（目前未使用） */
	uint16		flags;

	/* HASHED 阶段的字段 */
	uint8		nhashes;		/* 哈希函数数量 */
	uint32		nbits;			/* 位图中的位数（大小） */
	uint32		nbits_set;		/* 设置为 1 的位数 */

	/* 布隆过滤器的数据 */
	char		data[FLEXIBLE_ARRAY_MEMBER];
} BloomFilter;


/*
 * bloom_init
 * 		初始化布隆过滤器，分配所有内存。
 *
 * 过滤器的初始大小是针对预期的 ndistinct 值和
 * 请求的假阳性率的最佳大小。过滤器以 varlena 存储。
 */
static BloomFilter *
fc_bloom_init(int fc_ndistinct, double fc_false_positive_rate)
{
	Size		fc_len;
	BloomFilter *fc_filter;

	int			fc_nbits;			/* 过滤器的大小 / 位数 */
	int			fc_nbytes;			/* 过滤器的大小 / 字节数 */

	double		fc_k;				/* 哈希函数数量 */

	Assert(fc_ndistinct > 0);
	Assert(fc_false_positive_rate > 0 && fc_false_positive_rate < 1);

	/* 调整布隆过滤器大小：-(n * ln(p)) / (ln(2))^2 */
	fc_nbits = ceil(-(fc_ndistinct * log(fc_false_positive_rate)) / pow(log(2.0), 2));

	/* 将 m 四舍五入为整数字节 */
	fc_nbytes = ((fc_nbits + 7) / 8);
	fc_nbits = fc_nbytes * 8;

	/*
	 * 拒绝明显过大以至于无法存储在页面上的过滤器。
	 *
	 * 初始时布隆过滤器全为零，因此非常可压缩，但
	 * 随着值的增加，它变得越来越随机，因此压缩性也越来越差。
	 * 所以最初所有内容都能适应页面，但稍后可能会
	 * 遇到意外的失败——我们希望避免这种情况，因此拒绝
	 * 明显过大的布隆过滤器。
	 *
	 * XXX 有时会稍微过度设计布隆过滤器，以防止
	 * 意外的数据异常（某部分表的每个范围有更多的独特值等）。
	 * 但我们仍然需要确保即使是过大的过滤器也能适应页面，
	 * 如果有这样的需要的话。
	 *
	 * XXX 这个检查并不完美，因为索引可能有多个
	 * 过滤器，单独来看都比较小，但组合在一起时却太大。
	 */
	if (fc_nbytes > BloomMaxFilterSize)
		elog(ERROR, "the bloom filter is too large (%d > %zu)", fc_nbytes,
			 BloomMaxFilterSize);

	/*
	 * round(log(2.0) * m / ndistinct)，但假设在 Windows
	 * 上 round() 可能不可用
	 */
	fc_k = log(2.0) * fc_nbits / fc_ndistinct;
	fc_k = (fc_k - floor(fc_k) >= 0.5) ? ceil(fc_k) : floor(fc_k);

	/*
	 * 我们分配整个过滤器。大部分都是 0 位，因此
	 * varlena 容易压缩。
	 */
	fc_len = offsetof(BloomFilter, data) + fc_nbytes;

	fc_filter = (BloomFilter *) palloc0(fc_len);

	fc_filter->flags = 0;
	fc_filter->nhashes = (int) fc_k;
	fc_filter->nbits = fc_nbits;

	SET_VARSIZE(fc_filter, fc_len);

	return fc_filter;
}


/*
 * bloom_add_value
 * 		将值添加到布隆过滤器中。
 */
static BloomFilter *
fc_bloom_add_value(BloomFilter *fc_filter, uint32 fc_value, bool *fc_updated)
{
	int			fc_i;
	uint64		fc_h1,
				fc_h2;

	/* 计算哈希，用于布隆过滤器 */
	fc_h1 = hash_bytes_uint32_extended(fc_value, BLOOM_SEED_1) % fc_filter->nbits;
	fc_h2 = hash_bytes_uint32_extended(fc_value, BLOOM_SEED_2) % fc_filter->nbits;

	/* 计算请求的哈希数量 */
	for (fc_i = 0; fc_i < fc_filter->nhashes; fc_i++)
	{
		/* h1 + h2 + f(i) */
		uint32		fc_h = (fc_h1 + fc_i * fc_h2) % fc_filter->nbits;
		uint32		fc_byte = (fc_h / 8);
		uint32		fc_bit = (fc_h % 8);

		/* 如果位没有被设置，则设置它并记住我们这样做了 */
		if (!(fc_filter->data[fc_byte] & (0x01 << fc_bit)))
		{
			fc_filter->data[fc_byte] |= (0x01 << fc_bit);
			fc_filter->nbits_set++;
			if (fc_updated)
				*fc_updated = true;
		}
	}

	return fc_filter;
}


/*
 * bloom_contains_value
 * 		检查布隆过滤器是否包含特定值。
 */
static bool fc_bloom_contains_value(BloomFilter *fc_filter, uint32 fc_value)
{
	int			fc_i;
	uint64		fc_h1,
				fc_h2;

	/* 计算两个哈希值 */
	fc_h1 = hash_bytes_uint32_extended(fc_value, BLOOM_SEED_1) % fc_filter->nbits;
	fc_h2 = hash_bytes_uint32_extended(fc_value, BLOOM_SEED_2) % fc_filter->nbits;

	/* 计算请求的哈希数量 */
	for (fc_i = 0; fc_i < fc_filter->nhashes; fc_i++)
	{
		/* h1 + h2 + f(i) */
		uint32		fc_h = (fc_h1 + fc_i * fc_h2) % fc_filter->nbits;
		uint32		fc_byte = (fc_h / 8);
		uint32		fc_bit = (fc_h % 8);

		/* 如果位未设置，则值不存在 */
		if (!(fc_filter->data[fc_byte] & (0x01 << fc_bit)))
			return false;
	}

	/* 所有在布隆过滤器中找到的哈希值 */
	return true;
}

typedef struct BloomOpaque
{
	/*
	 * XXX 在这一点上我们只需要一个过程（计算哈希），但
	 * 为了一致性我们保留数组，就像包含和最小最大操作类一样。
	 * 将来可能还需要额外的过程。
	 */
	FmgrInfo	extra_procinfos[BLOOM_MAX_PROCNUMS];
	bool		extra_proc_missing[BLOOM_MAX_PROCNUMS];
} BloomOpaque;

static FmgrInfo *fc_bloom_get_procinfo(BrinDesc *fc_bdesc, uint16 fc_attno,
									uint16 fc_procnum);


Datum brin_bloom_opcinfo(PG_FUNCTION_ARGS)
{
	BrinOpcInfo *fc_result;

	/*
	 * opaque->strategy_procinfos 是惰性初始化的；这里由 palloc0 设置为
	 * all-uninitialized，fn_oid 设置为 InvalidOid。
	 *
	 * 布隆索引仅将过滤器作为单个 BYTEA 列存储
	 */

	fc_result = palloc0(MAXALIGN(SizeofBrinOpcInfo(1)) +
					 sizeof(BloomOpaque));
	fc_result->oi_nstored = 1;
	fc_result->oi_regular_nulls = true;
	fc_result->oi_opaque = (BloomOpaque *)
		MAXALIGN((char *) fc_result + SizeofBrinOpcInfo(1));
	fc_result->oi_typcache[0] = lookup_type_cache(PG_BRIN_BLOOM_SUMMARYOID, 0);

	PG_RETURN_POINTER(fc_result);
}

/*
 * brin_bloom_get_ndistinct
 *		确定用于设置布隆过滤器大小的 ndistinct 值。
 *
 * 根据 pagesPerRange 值调整 ndistinct 值。首先，
 * 如果它是负数，则假定相对于范围内的最大元组数（假设每页获得最大堆元组每页
 * 元组，这很可能是一个显著的高估）。我们还限制
 * 值，以避免不必要地增大布隆过滤器。
 *
 * XXX 只有在提供 pagesPerRange 值时我们才能做到这一点。
 * 如果没有，它必须是对索引的只读访问，在这种情况下我们不在乎。
 * 但也许我们应该回退到默认的 pagesPerRange 值？
 *
 * XXX 我们也可能获取列的 ndistinct 估算信息，
 * 并计算范围内预期的不同值数量。但由于数据以各种方式排序，
 * 所以依赖上限估计似乎更好。
 *
 * XXX 我们也可能计算每个 BRIN 范围的行数的更好估计，
 * 而不是使用 MaxHeapTuplesPerPage（这可能会产生远高于实际的值）。
 */
static int fc_brin_bloom_get_ndistinct(BrinDesc *fc_bdesc, BloomOptions *fc_opts)
{
	double		fc_ndistinct;
	double		fc_maxtuples;
	BlockNumber fc_pagesPerRange;

	fc_pagesPerRange = BrinGetPagesPerRange(fc_bdesc->bd_index);
	fc_ndistinct = BloomGetNDistinctPerRange(fc_opts);

	Assert(BlockNumberIsValid(fc_pagesPerRange));

	fc_maxtuples = MaxHeapTuplesPerPage * fc_pagesPerRange;

	/*
	 * 与 n_distinct 类似，负值是相对的 - 在这种情况下是相对于
	 * 页范围内的最大元组数（maxtuples）。
	 */
	if (fc_ndistinct < 0)
		fc_ndistinct = (-fc_ndistinct) * fc_maxtuples;

	/*
	 * 正值将被直接使用，但我们依然应用几个安全措施以避免使用不合理小的布隆过滤器。
	 */
	fc_ndistinct = Max(fc_ndistinct, BLOOM_MIN_NDISTINCT_PER_RANGE);

	/*
	 * 并且不要使用超出范围内可能的最大元组数，这将完全浪费。
	 */
	fc_ndistinct = Min(fc_ndistinct, fc_maxtuples);

	return (int) fc_ndistinct;
}

/*
 * 通过将给定的索引元组（包含某个页面范围的部分状态）与来自另一个堆
 * 元组的给定值进行比较来检查给定的索引元组。 如果新值超出了
 * 现有元组值指定的布隆过滤器，则更新索引元组并返回 true。
 * 否则，返回 false，并在这种情况下不进行修改。
 */
Datum brin_bloom_add_value(PG_FUNCTION_ARGS)
{
	BrinDesc   *fc_bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *fc_column = (BrinValues *) PG_GETARG_POINTER(1);
	Datum		fc_newval = PG_GETARG_DATUM(2);
	bool		fc_isnull PG_USED_FOR_ASSERTS_ONLY = PG_GETARG_DATUM(3);
	BloomOptions *fc_opts = (BloomOptions *) PG_GET_OPCLASS_OPTIONS();
	Oid			fc_colloid = PG_GET_COLLATION();
	FmgrInfo   *fc_hashFn;
	uint32		fc_hashValue;
	bool		fc_updated = false;
	AttrNumber	fc_attno;
	BloomFilter *fc_filter;

	Assert(!fc_isnull);

	fc_attno = fc_column->bv_attno;

	/*
	 * 如果这是第一个非空值，我们需要初始化布隆过滤器。
	 * 否则，仅从 BrinValues 中提取现有的布隆过滤器。
	 */
	if (fc_column->bv_allnulls)
	{
		fc_filter = fc_bloom_init(fc_brin_bloom_get_ndistinct(fc_bdesc, fc_opts),
							BloomGetFalsePositiveRate(fc_opts));
		fc_column->bv_values[0] = PointerGetDatum(fc_filter);
		fc_column->bv_allnulls = false;
		fc_updated = true;
	}
	else
		fc_filter = (BloomFilter *) PG_DETOAST_DATUM(fc_column->bv_values[0]);

	/*
	 * 使用提供的哈希函数计算新值的哈希，
	 * 然后将哈希值添加到布隆过滤器中。
	 */
	fc_hashFn = fc_bloom_get_procinfo(fc_bdesc, fc_attno, PROCNUM_HASH);

	fc_hashValue = DatumGetUInt32(FunctionCall1Coll(fc_hashFn, fc_colloid, fc_newval));

	fc_filter = fc_bloom_add_value(fc_filter, fc_hashValue, &fc_updated);

	fc_column->bv_values[0] = PointerGetDatum(fc_filter);

	PG_RETURN_BOOL(fc_updated);
}

/*
 * 给定一个与某个页面范围和扫描键对应的索引元组，返回扫描键是否与索引元组的布隆过滤器一致。若一致，则返回 true，否则返回 false。
 */
Datum brin_bloom_consistent(PG_FUNCTION_ARGS)
{
	BrinDesc   *fc_bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *fc_column = (BrinValues *) PG_GETARG_POINTER(1);
	ScanKey    *fc_keys = (ScanKey *) PG_GETARG_POINTER(2);
	int			fc_nkeys = PG_GETARG_INT32(3);
	Oid			fc_colloid = PG_GET_COLLATION();
	AttrNumber	fc_attno;
	Datum		fc_value;
	Datum		fc_matches;
	FmgrInfo   *fc_finfo;
	uint32		fc_hashValue;
	BloomFilter *fc_filter;
	int			fc_keyno;

	fc_filter = (BloomFilter *) PG_DETOAST_DATUM(fc_column->bv_values[0]);

	Assert(fc_filter);

	fc_matches = true;

	for (fc_keyno = 0; fc_keyno < fc_nkeys; fc_keyno++)
	{
		ScanKey		fc_key = fc_keys[fc_keyno];

		/* NULL键在bringetbitmap中处理并被过滤 */
		Assert(!(fc_key->sk_flags & SK_ISNULL));

		fc_attno = fc_key->sk_attno;
		fc_value = fc_key->sk_argument;

		switch (fc_key->sk_strategy)
		{
			case BloomEqualStrategyNumber:

				/*
				 * 在相等的情况下 (WHERE col = someval)，如果范围内的最小值 <= 扫描键，且最大值 >= 扫描键，我们想返回当前的页面范围。
				 */
				fc_finfo = fc_bloom_get_procinfo(fc_bdesc, fc_attno, PROCNUM_HASH);

				fc_hashValue = DatumGetUInt32(FunctionCall1Coll(fc_finfo, fc_colloid, fc_value));
				fc_matches &= fc_bloom_contains_value(fc_filter, fc_hashValue);

				break;
			default:
				/* 不应该发生 */
				elog(ERROR, "invalid strategy number %d", fc_key->sk_strategy);
				fc_matches = 0;
				break;
		}

		if (!fc_matches)
			break;
	}

	PG_RETURN_DATUM(fc_matches);
}

/*
 * 给定两个 BrinValues，将第一个更新为包含两个的摘要值的并集。第二个保持不变。
 *
 * XXX 我们假设布隆过滤器目前具有相同的参数。将来我们应有一个“可以合并”函数，以决定是否可以组合两个特定的布隆过滤器。
 */
Datum brin_bloom_union(PG_FUNCTION_ARGS)
{
	int			fc_i;
	int			fc_nbytes;
	BrinValues *fc_col_a = (BrinValues *) PG_GETARG_POINTER(1);
	BrinValues *fc_col_b = (BrinValues *) PG_GETARG_POINTER(2);
	BloomFilter *fc_filter_a;
	BloomFilter *fc_filter_b;

	Assert(fc_col_a->bv_attno == fc_col_b->bv_attno);
	Assert(!fc_col_a->bv_allnulls && !fc_col_b->bv_allnulls);

	fc_filter_a = (BloomFilter *) PG_DETOAST_DATUM(fc_col_a->bv_values[0]);
	fc_filter_b = (BloomFilter *) PG_DETOAST_DATUM(fc_col_b->bv_values[0]);

	/* 确保过滤器使用相同的参数 */
	Assert(fc_filter_a && fc_filter_b);
	Assert(fc_filter_a->nbits == fc_filter_b->nbits);
	Assert(fc_filter_a->nhashes == fc_filter_b->nhashes);
	Assert((fc_filter_a->nbits > 0) && (fc_filter_a->nbits % 8 == 0));

	fc_nbytes = (fc_filter_a->nbits) / 8;

	/* 简单地将位图进行 OR 操作 */
	for (fc_i = 0; fc_i < fc_nbytes; fc_i++)
		fc_filter_a->data[fc_i] |= fc_filter_b->data[fc_i];

	/* 更新过滤器中设置的位数 */
	fc_filter_a->nbits_set = pg_popcount((const char *) fc_filter_a->data, fc_nbytes);

	PG_RETURN_VOID();
}

/*
 * 缓存并返回包含操作类支持程序
 *
 * 返回与给定的函数支持编号对应的程序，如果不存在则返回 null。
 */
static FmgrInfo * fc_bloom_get_procinfo(BrinDesc *fc_bdesc, uint16 fc_attno, uint16 fc_procnum)
{
	BloomOpaque *fc_opaque;
	uint16		fc_basenum = fc_procnum - PROCNUM_BASE;

	/*
	 * 我们将其缓存于不透明结构中，以避免重复的系统缓存查找。
	 */
	fc_opaque = (BloomOpaque *) fc_bdesc->bd_info[fc_attno - 1]->oi_opaque;

	/*
	 * 如果我们已经搜索过这个过程且未找到它，就不必再搜索了。
	 */
	if (fc_opaque->extra_proc_missing[fc_basenum])
		return NULL;

	if (fc_opaque->extra_procinfos[fc_basenum].fn_oid == InvalidOid)
	{
		if (RegProcedureIsValid(index_getprocid(fc_bdesc->bd_index, fc_attno,
												fc_procnum)))
		{
			fmgr_info_copy(&fc_opaque->extra_procinfos[fc_basenum],
						   index_getprocinfo(fc_bdesc->bd_index, fc_attno, fc_procnum),
						   fc_bdesc->bd_context);
		}
		else
		{
			fc_opaque->extra_proc_missing[fc_basenum] = true;
			return NULL;
		}
	}

	return &fc_opaque->extra_procinfos[fc_basenum];
}

Datum brin_bloom_options(PG_FUNCTION_ARGS)
{
	local_relopts *fc_relopts = (local_relopts *) PG_GETARG_POINTER(0);

	init_local_reloptions(fc_relopts, sizeof(BloomOptions));

	add_local_real_reloption(fc_relopts, "n_distinct_per_range",
							 "number of distinct items expected in a BRIN page range",
							 BLOOM_DEFAULT_NDISTINCT_PER_RANGE,
							 -1.0, INT_MAX, offsetof(BloomOptions, nDistinctPerRange));

	add_local_real_reloption(fc_relopts, "false_positive_rate",
							 "desired false-positive rate for the bloom filters",
							 BLOOM_DEFAULT_FALSE_POSITIVE_RATE,
							 BLOOM_MIN_FALSE_POSITIVE_RATE,
							 BLOOM_MAX_FALSE_POSITIVE_RATE,
							 offsetof(BloomOptions, falsePositiveRate));

	PG_RETURN_VOID();
}

/*
 * brin_bloom_summary_in
 *		- brin_bloom_summary 类型的输入例程。
 *
 * brin_bloom_summary 仅用于内部表示 BRIN 布隆索引中的摘要，因此它没有自己的操作，我们也不允许输入。
 */
Datum brin_bloom_summary_in(PG_FUNCTION_ARGS)
{
	/*
	 * brin_bloom_summary 以二进制形式存储数据，不需要解析文本输入，因此不允许这样做。
	 */
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("cannot accept a value of type %s", "pg_brin_bloom_summary")));

	PG_RETURN_VOID();			/* 保持编译器安静 */
}


/*
 * brin_bloom_summary_out
 *		- brin_bloom_summary 类型的输出例程。
 *
 * BRIN 布隆摘要被序列化为 bytea 值，但我们希望输出一些人类可以理解的内容。
 */
Datum brin_bloom_summary_out(PG_FUNCTION_ARGS)
{
	BloomFilter *fc_filter;
	StringInfoData fc_str;

	/* 解压数据以获取带完整 4B 头的值 */
	fc_filter = (BloomFilter *) PG_DETOAST_DATUM(PG_GETARG_BYTEA_PP(0));

	initStringInfo(&fc_str);
	appendStringInfoChar(&fc_str, '{');

	appendStringInfo(&fc_str, "mode: hashed  nhashes: %u  nbits: %u  nbits_set: %u",
					 fc_filter->nhashes, fc_filter->nbits, fc_filter->nbits_set);

	appendStringInfoChar(&fc_str, '}');

	PG_RETURN_CSTRING(fc_str.data);
}

/*
 * brin_bloom_summary_recv
 *		- brin_bloom_summary 类型的二进制输入例程。
 */
Datum brin_bloom_summary_recv(PG_FUNCTION_ARGS)
{
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("cannot accept a value of type %s", "pg_brin_bloom_summary")));

	PG_RETURN_VOID();			/* 保持编译器安静 */
}

/*
 * brin_bloom_summary_send
 *		- brin_bloom_summary 类型的二进制输出例程。
 *
 * BRIN 布隆摘要被序列化为 bytea 值（尽管类型名称不同），所以我们就发送那个。
 */
Datum brin_bloom_summary_send(PG_FUNCTION_ARGS)
{
	return byteasend(fcinfo);
}
